package cc.verywell.pureblock.net;

import java.io.Closeable;
import java.lang.ref.WeakReference;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.function.Predicate;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.util.StringUtil;

/**
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月17日 上午10:19:38
*/
public class AioClientOuter implements Closeable,AioClientInterface<NormalSession, AioClientConnector>,CompletionHandler<NormalSession, AioClientOuter.IpCamp>,ConnectionFreeListener {
	public static InterruptedException STOPED_EXCEPTION = new InterruptedException("AioClinerOuter is Stopted");
	public static int EACH_PEER_REMOTES_MAX = 16; //每个peer最多只应该有16个地址;
	public static Function<Integer,Long> TimeOutFunction = x->(long)(5000+55000d/(x<1?1:x)); ///等待5-60秒
	
	public Predicate<SocketAddress> delayer = new AioClientDelayer();
	
	static boolean DEBUG = false;
	Runnable DEBUG_INIT = null;

	final AioClientInterface<NormalSession, AioClientInner.ClientConfig> clientInner;
	final ConnectionFreeProvider freeProvider;
	public AioClientOuter(AioClientInterface<NormalSession, AioClientInner.ClientConfig> clientInner, ConnectionFreeProvider freeProvider) {
		this.clientInner = clientInner;
		this.freeProvider = freeProvider;
		this.freeProvider.addConnectionFreeListener(this);
	}
	@Override
	public void close() {
		if(runstoped.compareAndSet(false, true)) {
			try {
				runlock.acquire();
				runlock.release();
				connectRemoteIp(null); //确保停止完毕;
			} catch (InterruptedException e) {
				if(Pblog.IO)Pblog.IO(e);
			} finally {
				freeProvider.removeConnectionFreeListener(this);
			}
		}
	}
	
	private Map<AioClientConnector,ValueCamp> values = new ConcurrentHashMap<>();
	private Map<String,IpCamp> ips = new ConcurrentHashMap<>();
	
	@Override
	public String toString() {
		return getClass().getSimpleName()+"("+Integer.toHexString(hashCode())+") ips:["+ips.size()+"], remote:["+values.size()+"], running:["+runingList.size()+"], finish:["+finishList.size()+"], runstack:["+runstack.size()+"]";
	}
	@Override
	public <A> Future<NormalSession> connect(AioClientConnector remote, A attachment, CompletionHandler<NormalSession, A> handler) {
		Objects.requireNonNull(remote);
		
		Future<NormalSession> future = null;
		ValueCamp vcnew,vc = vcnew = new ValueCamp(remote);
		vc:while(vc!=null) {
			synchronized (vc) {
				if(vc.done) {
					vc = vcnew = new ValueCamp(remote);
					continue vc;
				}
				ValueCamp oldvc = values.putIfAbsent(remote, vc);
				if(oldvc!=null && oldvc!=vc) {
					vc = oldvc;
					continue vc;
				}
				if(vc == vcnew)vc.init();
				/////已确保dw锁定并唯一;
				future = vc.connect(attachment, handler);
				/////离开vc锁定并唯一;
				break vc;
			}
		}
		
		//先检查一下看看对方是不是已经主动连接过来了;
		NormalSession session = remote.getSession();
		if(session!=null && !session.isClosed()) {
			if(Pblog.IO)Pblog.IO("AioClient ",remote.getRemote()," is connected at ",session);
			vc.completed(session, null);
			return future;
		}
		
		//新建立ValueCamp时要先把连接地址分好上报;
		IpCamp recommandIp = null;
		if(vc == vcnew) {
			
			String[] remotestrs = remote.getRemote();
			Map<String,IpCamp> waitJoinAl;
			if(remotestrs!=null) {
				int i = 0;
				waitJoinAl = new HashMap<String,IpCamp>(remotestrs.length);
				for (String remotestr : remotestrs) {
					if(remotestr==null)continue;
					if(waitJoinAl.containsKey(remotestr))continue; //避免重复bind,ipnew,会造成ip长期留传(只能通过连接删除);
					if(EACH_PEER_REMOTES_MAX<++i)break;
					
					IpCamp ipnew,ip = ipnew = new IpCamp(remotestr);
					ip:while(ip!=null) {
						synchronized (ip) {
							if(ip.done) {
								ip = ipnew = new IpCamp(remotestr);
								continue ip;
							}
							IpCamp oldip = ips.putIfAbsent(remotestr, ip);
							if(oldip!=null && oldip!=ip) {
								ip = oldip;
								continue ip;
							}
							if(ip == ipnew)ip.init();
							/////已确保vc锁定并唯一;
							ip.bind(vc);
							if(ip.future==null) { //如果全部ip都是连接中的，不会释放准连;
								if(recommandIp==null)recommandIp = ip;
							}
							waitJoinAl.put(remotestr,ip);
							/////离开vc锁定并唯一;
							break ip;
						}
					}
				}
			}else {
				waitJoinAl = Collections.EMPTY_MAP;
			}
			
			if(DEBUG_INIT!=null)DEBUG_INIT.run();
			
			if(vc.initIp(waitJoinAl.values())) { //waitJoinAl可以为empty, 到这里会直接输出结果;
				return future;
			}
			
		}
		
		//////进入连接进程
		{
			if(recommandIp!=null)finishList.add(Collections.singletonList(new WeakReference<>(vc))); //到这里才释放准入;
			connectRemoteIp(recommandIp);
		}

		return future;
	}
	
	//下载中同步器////////////////////////////////////////////////////////////////////
	
	final private Semaphore runlock = new Semaphore(1); ///用来处理所有新建连接不会给重复调用;
	final private BlockingQueue<IpCamp> runstack = new LinkedBlockingQueue<>();
	final private AtomicBoolean runstoped = new AtomicBoolean(false);
	final private BlockingQueue<List<WeakReference<ValueCamp>>> finishList = new LinkedBlockingQueue<>();
	final private LinkedList<WeakReference<IpCamp>> runingList = new LinkedList<>(); //只能在runlock线程内调用;
	private void connectRemoteIp(IpCamp recommand) {
		if(recommand!=null)runstack.offer(recommand);
		if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.lock_try ",Thread.currentThread());
		int recommandcount = 0;
		boolean tempPass;
		connectloop:do{
			tempPass = false;
			if(runlock.tryAcquire()) {
				try {
					if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.lock_get ",Thread.currentThread());
					if(runstoped.get()) {
						//停止所有进行中的链接
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.stoping [",ips.size(),"]");
						Iterator<IpCamp> it = ips.values().iterator();
						while(it.hasNext()) {
							IpCamp ic = it.next();
							if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.stoping ",ic);
							ic.cancel(STOPED_EXCEPTION);
							it.remove();
						}
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.stoping remain ips:[",ips.size(),"], remote:[",values.size(),"]");
						runstack.clear();
						//finishList.clear();
						return;
					}else {
						//连接后数据清理
						List<WeakReference<ValueCamp>> ls = null;
						while(null!=(ls=finishList.poll())) {
							for (WeakReference<ValueCamp> r : ls) {
								ValueCamp v = r.get();
								if(v!=null) {
									v.downingSync--;
									if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run unmark ",v,"=",v.downingSync);
								}
							}
						}
						//检查超时进程
						int timeoutSize;
						if(Pblog.DEBUG && DEBUG)timeoutSize = runingList.size();else timeoutSize=0;
						WeakReference<IpCamp> w;
						while(null!=(w=runingList.peek())){
							IpCamp ic = w.get();
							if(ic==null) {
								runingList.poll();
								continue;
							}
							synchronized (ic) {
								if(ic.done) {
									runingList.poll();
									continue;
								}
								long now = System.currentTimeMillis();
								long expire = now-TimeOutFunction.apply(runingList.size());
								if(expire<ic.connectTime && ic.connectTime<=now) {
									break; ////往后的时间只会比他更大
								}
							}
							if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.timeout ",ic);
							runingList.poll();
							{
								AioClientOuter.this.ips.remove(ic.address, ic);
								ic.cancel(new TimeoutException());
							}
						}
						if(Pblog.DEBUG && DEBUG && timeoutSize!=0)Pblog.DEBUG("AioClient.run.timeout [",timeoutSize,"->",runingList.size(),"] ");

					}
					if(freeProvider.isTooManyPendingConnections()) {
						return;
					}
					IpCamp ipcheck = runstack.poll();
					if(null!=ipcheck) {
						Throwable err = null;
						//建立地址
						if(err==null && ipcheck.getRemoteAddress()==null){
							try {
								InetSocketAddress ip = StringUtil.stringToPort(ipcheck.address); ///这里会耗费几秒;
								ipcheck.setRemoteAddress(ip); 
								//TODO 这里应该加入地址预检查;
							} catch (IllegalArgumentException e) {
								if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.abenden ",ipcheck," <",e,">");
								//e.printStackTrace();
								err = e.fillInStackTrace();
							}
						}
						//重连间隔名单在此过滤,只要在名单内直接删除
						if(err==null && delayer!=null
							&& delayer.test(ipcheck.getRemoteAddress())) {
								synchronized (ipcheck) {
									if(ipcheck.future==null) {
										if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.abenden ",ipcheck," <.delayer.>");
										err = new TimeoutException();
									}
								}
						}
						if(err!=null) {
							AioClientOuter.this.ips.remove(ipcheck.address, ipcheck);
							ipcheck.cancel(err);
						}else {
							//检查看看IpCamp是否需要链接;
							synchronized (ipcheck) {
								boolean pass = !ipcheck.done && ipcheck.future==null;
								pass:if(pass) {
									for (ValueCamp vc : ipcheck.valueal) {
										if(vc.downingSync==0) {
											break pass;
										}
									}
									if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.abenden ",ipcheck," <.downing.>");
									pass = false;
								}
								if(pass) {
									if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.connect->: ",ipcheck," linked [",ipcheck.valueal.size(),"]");
									ArrayList<WeakReference<ValueCamp>> downList = new ArrayList<WeakReference<ValueCamp>>(ipcheck.valueal.size());
									ipcheck.downingSync = downList;
									ipcheck.connectTime = System.currentTimeMillis();
									runingList.add(new WeakReference<>(ipcheck));
									for (ValueCamp vc : ipcheck.valueal) {
										vc.downingSync++;
										downList.add(new WeakReference<AioClientOuter.ValueCamp>(vc));
										if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run mark ",vc,"=",vc.downingSync);
									}
									///留到最后，因为在connet同时可能就会返回false 或者 completed
									Future<NormalSession> f = clientInner.connect(ipcheck, ipcheck, this);
									ipcheck.putFuture(f);
									if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.connect<-: ",ipcheck," future:",f);
									tempPass = true;//因为会有即时返回的情况,应该先处理完【连接后数据清理】，再【发掘是否有需要连接的内容】
									continue connectloop;
								}
							}
						}
					}
					if(null==runstack.peek()){
						//发掘是否有需要连接的内容，是就加入runstack;
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.recommand remain ips:[",ips.size(),"], remote:[",values.size(),"]");
						//ValueCamp result = null; //需求futures要多，加入时间initTime要小
						long minInitTime = Long.MAX_VALUE;
						int maxFutures = 0;
						List<IpCamp> result = null;
						for(ValueCamp vc : values.values()) {
							if(vc.downingSync==0) {
								synchronized (vc) {
									if(vc.downingSync==0 && !vc.done && vc.initTime!=0) {
										if(vc.futures<maxFutures)continue;
										if(vc.futures>maxFutures || vc.initTime<minInitTime) {
											//result = vc;
											result = vc.ipmap.getIps();
											maxFutures = vc.futures;
											minInitTime = vc.initTime;
										}
									}
								}
							}
						}
						if(result!=null) {
							int ipRelations = 0;
							IpCamp ip = null;
							for(IpCamp c : result) {
								synchronized (c) {
									if(!c.done && c.future==null) {
										if(c.valueal.size()>ipRelations) {
											ipRelations = c.valueal.size();
											ip = c;
										}
									}else {
										if(Pblog.IO)Pblog.DEBUG("AioClient.run.recommand Finded a illegal IpCamp downingSync:",c.downingSync!=null," done:",c.done," future:",c.future);
									}
								}
							}
							if(ip!=null) {
								runstack.offer(ip);
							}else {
								recommandcount++;
								if(Pblog.DEBUG)Pblog.DEBUG("AioClient.run.recommand Finded a unUseful remote",recommandcount,": ",result);// 应该是临时被使用了吧。
								if(recommandcount>5)new IllegalAccessException().printStackTrace();
								else tempPass = true;
							}
							if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.recommand new ",ip," in ",result);
						}else {
							if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.recommand [no]");
						}
					}
				}finally {
					runlock.release();
					if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.lock_free ",Thread.currentThread());
				}
			}else {
				if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.lock_miss ",Thread.currentThread());
				return;
			}
		}while(tempPass || null!=runstack.peek());
	}
	@Override
	public void connectionfree() {
		if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.run.connectionFree ",Thread.currentThread());
		connectRemoteIp(null);
	}
	
	//对网络inner包/////////////////////////////////////////////////////////////////////
	static public class IpCamp implements AioClientInner.ClientConfig{
		public List<WeakReference<ValueCamp>> downingSync = null; //只能在runlock线程内调用,本地无调用权限;
		public long connectTime = -1; //只能在runlock线程内调用,本地无调用权限;
		
		final String address;
		LinkedList<ValueCamp> valueal;
		public IpCamp(String address) {
			this.address = address;
		}
		public void init() {
			valueal = new LinkedList<>();
		}
		
		////ClientConfig功能区 - AioClientInner.ClientConfig
		private SocketAddress remoteAddress;
		public void setRemoteAddress(SocketAddress ip) {
			this.remoteAddress = ip;
		}
		@Override
		public SocketAddress getRemoteAddress() {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp.asked:RemoteAddress ",this,"- ",remoteAddress,"  ",Thread.currentThread());
			return remoteAddress;
		}
		private SocketAddress bindAddress;
		public void setLocalBindAddress(SocketAddress ip) {
			this.remoteAddress = ip;
		}
		@Override
		public SocketAddress getLocalBindAddress() {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp.asked:BindAddress ",this,".",bindAddress,"  ",Thread.currentThread());
			return bindAddress;
		}
		@Override
		public void beforeConnect(AsynchronousSocketChannel channel) {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp.asked:beforeConnect ",this,"  ",Thread.currentThread());
			//添加连接参数
		}
		
		
		public boolean done = false;
		private Future<NormalSession> future;
		public synchronized void putFuture(Future<NormalSession> future) {
			//if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp.future<-: ",this,".(",future,") ",Thread.currentThread());
			if(this.future!=null)throw new IllegalAccessError();
			this.future = future;
		}

		public synchronized void bind(ValueCamp vc) {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp[v].bind: ",vc.remote,"-",this);
			if(done)throw new IllegalAccessError();
			valueal.add(vc);
		}
		
		///流程入口 3/3  //必须和ips.remove配套使用
		public void cancel(Throwable exc){
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp[x](3/3).cancel->finish ",this,"  ",Thread.currentThread());
			LinkedList<ValueCamp> val = finish();
			if(val!=null) { //如果为null即表明已done
				for (ValueCamp vc : val) {
					vc.failed(exc, this);
				}
				if(future!=null)future.cancel(false);
			}

		}
		
		////流程入口 1/3  //必须和ips.remove配套使用
		public synchronized LinkedList<ValueCamp> finish(){
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp[x](1/3).finish ",valueal,"-",this,"  ",Thread.currentThread());
				if(done) {
					if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp[x](1/3).finished before ",this,"  ",Thread.currentThread());
					return null;
				}
				done = true;
				LinkedList<ValueCamp> val = valueal;
				valueal = null;
				//等会要用future = null;
				return val;
		}
		
		
		////流程入口 2/3
		public boolean free(ValueCamp vc) {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.IpCamp[x].unbind: ",vc.remote,"-",this," (2/3)");
			Future<NormalSession> f = null;
			boolean removeFromMap = false;
			synchronized (this) {
				if(done)return false;
				if(vc!=null)valueal.remove(vc);
				if(valueal.isEmpty()) {
					done = true;
					removeFromMap = true;
					valueal = null;
//					if(future!=null) {
						f = future;
//						future = null;
//					}
				}
			}
			if(f!=null) {
				f.cancel(false);
			}
			return removeFromMap;
		}
		
		@Override
		public String toString() {
			return address;
		}
	}
	
	@Override
	public void completed(NormalSession ns, IpCamp ic) {
		if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.completed ",ic,"(",ns,") ",Thread.currentThread());
		AioClientOuter.this.finishList.add(ic.downingSync);
		AioClientOuter.this.ips.remove(ic.address, ic);
		LinkedList<ValueCamp> val = ic.finish();
		if(val!=null) { //发生于AioClient.IpCamp[x](3/3).cancel, 然后在IpCamp.unbind过程中Inner完成了
			for (ValueCamp vc : val) {
				vc.completed(ns, ic);
			}
		}
	}
	@Override
	public void failed(Throwable exc, IpCamp ic) {
		if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.failed ",ic," <",exc,"> ",Thread.currentThread());
		AioClientOuter.this.finishList.add(ic.downingSync);
		AioClientOuter.this.ips.remove(ic.address, ic);
		LinkedList<ValueCamp> val = ic.finish();
		if(val!=null) { //发生于AioClient.ValueCamp[x](3/4).cancelFuture.DONE, 因为在IpCamp.unbind过程中也会Inner.cancelFuture
			for (ValueCamp vc : val) {
				vc.failed(exc, ic);
			}
		}
	}

	
	///对外Outer包////////////////////////////////////////////////////////////////////
	
	private class ValueCamp{
		public int downingSync = 1; //只能在runlock线程内调用,本地无调用权限;, 首次运行需在finishList释放准连
		
		final public AioClientConnector remote;
		public LinkedList<AttachmentCamp<?>> todoal;
		public ValueCampIpJob ipmap;
		public ValueCamp(AioClientConnector remote) {
			this.remote = remote;
		}
		public void init() {
			todoal = new LinkedList<>();
			ipmap = new ValueCampIpJob();
		}
		
		@Override
		public String toString() {
			return remote+".value("+Integer.toHexString(ValueCamp.this.hashCode())+")"+(done?"done":"");
		}
		
		public boolean done = false;
		public int futures = 0;
		public NormalSession normalSession;
		public long initTime = 0;
		
		///流程入口 4/4
		public boolean initIp(Collection<IpCamp> inital) {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[v](4/4).initIp ",remote,"-[",inital.size(),"] ",Thread.currentThread());
			LinkedList<AttachmentCamp<?>> tal = null;
			Collection<IpCamp> ial;
			Throwable failedexc;
			boolean isDone = false;
			synchronized (ValueCamp.this) {
				if(ipmap.initCauseDone(inital)) {
					if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](4/4).initIp.DONE ",remote,"-[",ipmap.getSize(),"]",Thread.currentThread());
					if(done)throw new IllegalAccessError();
					done = true;
					tal = todoal;
					todoal = null;
					failedexc = ipmap.failedExc();
					ial = ipmap.initFreeList(); //一定是null;
					//ipmap = null;
					ValueCamp.this.notifyAll(); //释放所有Future
					AioClientOuter.this.values.remove(remote, ValueCamp.this);
				}else {
					failedexc = ipmap.failedExc(); //拿了没用
					ial = ipmap.initFreeList();
				}
				isDone = done;
				initTime = System.currentTimeMillis();
			}
			if(ial!=null) {
				for (IpCamp ic : ial) {
					if(ic.free(ValueCamp.this)) {
						AioClientOuter.this.ips.remove(ic.address, ic);
					}
				}
			}
			if(tal!=null) {
				AttachmentCamp<?> ac ;
				while(null!=(ac=tal.poll())) {
					ac.failed(failedexc); //cancel所有handel
				}
			}
			return isDone;
		}
		
		////流程入口 1/4
		public void completed(NormalSession ns,IpCamp ipcamp) {
			//if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](1/4).completed ",remote,"-",ipcamp,"(",ns,") ",Thread.currentThread());
			if(remote.getSession()!=ns) {
				failed(null, ipcamp);
				return;
			}
			LinkedList<AttachmentCamp<?>> tal = null;
			Collection<IpCamp> ial = null;
			synchronized (ValueCamp.this) {
				if(done)return;
				if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](1/4).completed.DONE ",remote,"-[",ipmap.getSize(),"]",ipcamp,"(",ns,") ");
				normalSession = ns;
				done = true;
				
				tal = todoal;
				todoal = null;
				ial = ipmap.completedAndDoneAndFreeList(ipcamp);
				//ipmap = null;
				ValueCamp.this.notifyAll();//释放所有Future
				AioClientOuter.this.values.remove(remote, ValueCamp.this);
			}
			if(ial!=null) {
				for (IpCamp ic : ial) {
					if(ic.free(ValueCamp.this)) {
						AioClientOuter.this.ips.remove(ic.address, ic);
					}
				}
			}
			if(tal!=null) {
				AttachmentCamp<?> ac ;
				while(null!=(ac=tal.poll())) {
					ac.completed(ns); //finish所有handel
				}
			}
		}
		////流程入口 2/4
		public void failed(Throwable exc,IpCamp ipcamp) {
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](2/4).failed ",remote,"-",ipcamp," <",exc,"> ",Thread.currentThread());
			LinkedList<AttachmentCamp<?>> tal = null;
			//Collection<IpCamp> ial = null;
			synchronized (ValueCamp.this) {
				if(done)return;
				//ipmap.failedCauseFree(ipcamp);
				if(ipmap.failedCauseDone(ipcamp,exc)) {
					if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](2/4).failed.DONE ",remote,"-[",ipmap.getSize(),"]");
					done = true;
					tal = todoal;
					todoal = null;
					//ipmap已空ial = ipmap.values();
					//ipmap = null;
					ValueCamp.this.notifyAll(); //释放所有Future
					AioClientOuter.this.values.remove(remote, ValueCamp.this);
				}
			}
			//if(ial!=null) {
			//	for (IpCamp ic : ial) {
			//		if(ic.free(ValueCamp.this)) {
			//			AioClientOuter.this.ips.remove(ic.address, ic);
			//		}
			//	}
			//}
			if(tal!=null) {
				AttachmentCamp<?> ac ;
				while(null!=(ac=tal.poll())) {
					ac.failed(exc); //cancel所有handel
				}
			}
		}
		//void futureCancel(AttachmentCamp<?> attachmentCamp) {
		////内容已移动到OuterFuture.cancel;	
		//}
		////
		public synchronized <A> Future<NormalSession> connect(A attachment, CompletionHandler<NormalSession, A> handler){
			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp.newFuture ",remote,"  ",Thread.currentThread());
			if(done)throw new IllegalAccessError();
			AttachmentCamp<A> ac;
			if(handler!=null) {
				ac = new AttachmentCamp<>(attachment,handler);
				todoal.add(ac);
			}else ac = null;
			return new OuterFuture(ac);
		}
		private class OuterFuture implements Future<NormalSession>{
			private AttachmentCamp<?> attachmentCamp;
			boolean isCancelled = false;
			OuterFuture(AttachmentCamp<?> attachmentCamp) { ///必须在 synchronized (ValueCamp.this) 内运行;
				this.attachmentCamp = attachmentCamp;
				futures++;
			}
			
			////Future功能区 - Future<NormalSession>
			@Override
			////流程入口3/4
			public boolean cancel(boolean mayInterruptIfRunning) {
				if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp(3/4).cancelFuture ",remote,"  ",Thread.currentThread());
				AttachmentCamp<?> ac = null;
				LinkedList<AttachmentCamp<?>> tal = null;
				Collection<IpCamp> ial = null;
				synchronized (ValueCamp.this) {
					if(isCancelled || done)return false;
					isCancelled = true;
					if(attachmentCamp!=null) {
						ac = attachmentCamp;
						todoal.remove(attachmentCamp);
						attachmentCamp = null;
					}
					futures--;
					ValueCamp.this.notifyAll();//释放所有Future
					if(futures<=0) {
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp[x](3/4).cancelFuture.DONE ",remote,"-[",ipmap.getSize(),"]");
						done = true;
						tal = todoal;
						todoal = null;
						ial = ipmap.cancelAndDoneAndFreeList();
						//ipmap = null;
						values.remove(remote, ValueCamp.this);
					}
				}
				if(ial!=null) {
					for (IpCamp ic : ial) {
						if(ic.free(ValueCamp.this)) {
							AioClientOuter.this.ips.remove(ic.address, ic);
						}
					}
				}
				if(ac!=null)ac.failed(null);//cancel自己的handel
				if(tal!=null) {
					while(null!=(ac=tal.poll())) {
						ac.failed(null);//cancel所有handel
					}
				}
				return true;
			}
			@Override
			public boolean isCancelled() {
				return isCancelled;
			}

			@Override
			public boolean isDone() {
				return isCancelled || done;
			}

			@Override
			public NormalSession get() throws InterruptedException, ExecutionException {
				synchronized (ValueCamp.this) {
					while(!done && !isCancelled) {
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp.Future.waiting ",remote,"  ",Thread.currentThread());
						ValueCamp.this.wait();
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp.Future.wakeup ",remote,"  ",Thread.currentThread());
					}
					return isCancelled?null:normalSession;
				}
			}

			@Override
			public NormalSession get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
				timeout = TimeUnit.MILLISECONDS.convert(timeout,unit);
				long dest = timeout+System.currentTimeMillis();
				synchronized (ValueCamp.this) {
					{
						long rest = dest-System.currentTimeMillis();
						if(rest>timeout)timeout = -1; //时钟出现了问题;
						else {
							if(rest==timeout)rest--;
						 	timeout = rest;
						}
					}
					while(timeout>0) {
						if(isCancelled)return null;
						if(done)return normalSession;
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp.Future.waiting ",remote,"  ",Thread.currentThread()," time: ",timeout);
						ValueCamp.this.wait(timeout);
						if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient.ValueCamp.Future.wakeup ",remote,"  ",Thread.currentThread());
						{
							long rest = dest-System.currentTimeMillis();
							if(rest>timeout)timeout = -1; //时钟出现了问题;
							else {
								if(rest==timeout)rest--;
							 	timeout = rest;
							}
						}
					}
					if(isCancelled)return null;
					if(done)return normalSession;
					else throw new TimeoutException();
				}
				
			}
			
			@Override
			public String toString() {
				return remote+".value("+Integer.toHexString(ValueCamp.this.hashCode())+")"+(done?"done":"")+".future("+Integer.toHexString(hashCode())+")"+(isCancelled?"cancelled":"");
			}

		}

	}
	static private class ValueCampIpJob{ //所有内容的修改，访问，都必须在 syncronizd ValueCamp内进行;
		boolean isInit = false;
		boolean weDone = false;
		boolean outerDone = false;
		private LinkedList<IpCamp> movedLink;
		public Map<String,IpCamp> ipmap;
		public List<IpCamp> getIps(){
			return new ArrayList<>(ipmap.values());
		}
		public int getSize() {
			if(isInit) {
				if(ipmap!=null)return ipmap.size();
				else return 0;
			}else {
				if(movedLink!=null)return 0-movedLink.size();
				else return 0;
			}
		}
		//对外接口 4/4
		public boolean initCauseDone(Collection<IpCamp> al) {
//			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient ValueCampIpJob(4/4) initCauseDone ",al.size());
			Objects.requireNonNull(al);
			if(weDone || isInit)throw new IllegalAccessError();
			isInit = true;
			if(outerDone) {
				if(movedLink!=null) {
					ArrayList<IpCamp> ipal = new ArrayList<IpCamp>(al.size());
					for (IpCamp ip : al) {
						if(!movedLink.remove(ip)) {
							ipal.add(ip);
						}
					}
					movedLink = null;
					al = ipal;
				}
				weDone = true;
				initFreeList = al;
				return false;
			}else{
				ipmap = new HashMap<>(al.size());
				if(movedLink!=null) {
					for (IpCamp ip : al) {
						if(!movedLink.remove(ip)) {
							ipmap.put(ip.address, ip);
						}
					}
					movedLink = null;
				}else {
					for (IpCamp ip : al) {
						ipmap.put(ip.address, ip);
					}
				}
				if(ipmap.isEmpty()) {
					weDone = outerDone = true;
					ipmap = null;
					return true;
				}else {
					return false;
				}
			}
		}
		////如果上面返回true，下面两个会被调用
		private Collection<IpCamp> initFreeList;
		private Throwable failedexc;
		Collection<IpCamp> initFreeList() {
			Collection<IpCamp> r = initFreeList;
			initFreeList = null;
			return r;
		}
		Throwable failedExc() {
			Throwable r = failedexc;
			failedexc = null;
			return r;
		}
		
		//对外接口 2/4
		boolean failedCauseDone(IpCamp ipcamp,Throwable failedexc) { //下达了failed通知
//			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient ValueCampIpJob(2/4) failedCauseDone");
			if(weDone || outerDone)throw new IllegalAccessError();
			if(isInit) {
				if(ipcamp!=null)
					ipmap.remove(ipcamp.address, ipcamp);
				if(ipmap.isEmpty()) {
					weDone = outerDone = true;
					ipmap = null;
					return true;
				}else {
					return false;
				}
			}else {
				if(failedexc!=null)this.failedexc = failedexc;
				if(movedLink==null)
					movedLink = new LinkedList<>();
				movedLink.add(ipcamp);
				return false;
			}
		}
		///对外接口 1/4
		Collection<IpCamp> completedAndDoneAndFreeList(IpCamp ipcamp) { //下达了completed通知
//			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient ValueCampIpJob(1/4) completedAndDoneAndFreeList");
			if(weDone || outerDone)throw new IllegalAccessError();
			outerDone = true;
			if(isInit) {
				weDone = true;
				if(ipcamp!=null)
					ipmap.remove(ipcamp.address, ipcamp);
				Collection<IpCamp> al = ipmap.values();
				ipmap = null;
				return al;
			}else {
				if(ipcamp!=null) {
					if(movedLink==null)
						movedLink = new LinkedList<>();
					movedLink.add(ipcamp);
				}
				return null;
			}
		}
		//对外接口 3/4
		Collection<IpCamp> cancelAndDoneAndFreeList() { //下达了cancel通知;
//			if(Pblog.DEBUG && DEBUG)Pblog.DEBUG("AioClient ValueCampIpJob(3/4) cancelAndDoneAndFreeList");
			return completedAndDoneAndFreeList(null);
		}
			
	}
	static class AttachmentCamp<A>{
		final A attachment;
		final CompletionHandler<NormalSession, A> handler;
		public AttachmentCamp(A attachment, CompletionHandler<NormalSession, A> handler) {
			this.attachment = attachment;
			this.handler = handler;
		}
		public void completed(NormalSession normalsession) {
			this.handler.completed(normalsession, attachment);
		}
		public void failed(Throwable exc) {
			this.handler.failed(exc, attachment);
		}

	}

	
	
}
